/*
 * @(#)ChannelManage.java   1.0  2021年11月8日
 * 
 * Copyright (c)	2014-2020. All Rights Reserved.	GuangZhou hhmk Technology Company LTD.
 */
package com.swift.dts.channel;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.swift.core.env.ConfigInterface;
import com.swift.dts.filter.DtsFilter;
import com.swift.dts.loadconfig.model.DtsDataBase;
import com.swift.dts.model.DtsBinlogMessage;
import com.swift.dts.sink.ExecSqlSink;
import com.swift.dts.util.CreateExecSql;
import com.swift.util.type.TypeUtil;

/**
 * 处理管道流程控制器
 * @author zhengjiajin
 * @version 1.0 2021年11月8日
 */
@Component
public class ChannelManage {

    @Autowired(required=false)
    private List<DtsFilter> filterList;

    @Autowired
    private ConfigInterface<DtsDataBase> dataBaseConfigInterface;
    
    @Autowired
    private ExecSqlSink execSqlSink;
    
    public void listener(DtsBinlogMessage message) {
        List<DtsDataBase> dbList = getDestinationDb();
        if(TypeUtil.isNull(dbList)) return;
        CreateExecSql.setExecSql(message);
        if(dbList.size()<=1) {
            runOne(dbList.get(0), message);
        }else {
            runMore(dbList, message);
        }
    }
    
    private void runOne(DtsDataBase todb,DtsBinlogMessage message) {
        if(filterList!=null) {
            for(DtsFilter filter:filterList) {
                if(filter.filter(todb,message)) return;
            }
        }
        execSqlSink.execSql(todb,message);
    }
    
    private void runMore(List<DtsDataBase> dbList,DtsBinlogMessage message) {
        final CountDownLatch countDownLatch = new CountDownLatch(dbList.size());
        for(DtsDataBase db:dbList) {
            new Thread(() -> {
                try {
                    if(filterList!=null) {
                        for(DtsFilter filter:filterList) {
                            if(filter.filter(db,message)) return;
                        }
                    }
                    execSqlSink.execSql(db,message);
                }finally {
                    countDownLatch.countDown();
                }
            }).start();
        }
        try {
            countDownLatch.await(30,TimeUnit.SECONDS);
        } catch (InterruptedException e) {
        }
    }
    
    public List<DtsDataBase> getDestinationDb(){
        List<DtsDataBase> destinationDbList = new LinkedList<DtsDataBase>();
        List<DtsDataBase> dbList = dataBaseConfigInterface.load();
        if(TypeUtil.isNull(dbList)) return null;
        for(DtsDataBase db:dbList) {
            if(!db.getIsSource()) destinationDbList.add(db);
        }
        return destinationDbList;
    }

}
